#include "service/DataUploader.h"
#include "core/Logger.h"
#include "utils/Config.h"
#include "utils/HardwareInfo.h"
#include "protocol/mqtt/MqttClient.h"
#include "device/DeviceManager.h"

#include <openssl/hmac.h>
#include <openssl/evp.h>

#include <atomic>
#include <chrono>
#include <thread>
#include <fstream>
#include <sstream>
#include <algorithm>
#include <cctype>
#include <ctime>
#include <cstdlib>
#include <cstdio>
#include <unordered_set>

namespace ems {
namespace service {

namespace {

std::string toLower(std::string s) {
    std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c){ return static_cast<char>(std::tolower(c)); });
    return s;
}

std::string replaceChar(std::string s, char from, char to) {
    std::replace(s.begin(), s.end(), from, to);
    return s;
}

std::string sanitizeChannelToken(const std::string& value) {
    // 将地址/从站号等片段规范为小写字母数字与连字符，便于生成通道名
    std::string out;
    out.reserve(value.size());
    bool lastDash = false;
    for (char ch : value) {
        unsigned char uc = static_cast<unsigned char>(ch);
        if (std::isalnum(uc)) {
            out.push_back(static_cast<char>(std::tolower(uc)));
            lastDash = false;
        } else if (!out.empty() && !lastDash) {
            out.push_back('-');
            lastDash = true;
        }
    }
    if (!out.empty() && out.back() == '-') {
        out.pop_back();
    }
    return out;
}

bool isDigits(const std::string& s) {
    return !s.empty() && std::all_of(s.begin(), s.end(), [](unsigned char ch){ return std::isdigit(ch); });
}

std::string extractIpLastOctet(const std::string& address) {
    auto trim = [](const std::string& s) -> std::string {
        const auto begin = std::find_if_not(s.begin(), s.end(), [](unsigned char ch){ return std::isspace(ch); });
        if (begin == s.end()) {
            return std::string();
        }
        const auto end = std::find_if_not(s.rbegin(), s.rend(), [](unsigned char ch){ return std::isspace(ch); }).base();
        return std::string(begin, end);
    };

    std::string normalized = trim(address);
    auto colonPos = normalized.find(':');
    if (colonPos != std::string::npos) {
        normalized = normalized.substr(0, colonPos);
    }
    size_t lastDot = normalized.find_last_of('.');
    if (lastDot == std::string::npos || lastDot + 1 >= normalized.size()) {
        return {};
    }

    std::string lastSegment = normalized.substr(lastDot + 1);
    if (!isDigits(lastSegment)) {
        return {};
    }

    bool valid = true;
    size_t start = 0;
    int segments = 0;
    while (valid && start < lastDot) {
        size_t dotPos = normalized.find('.', start);
        if (dotPos == std::string::npos || dotPos > lastDot) {
            break;
        }
        std::string segment = normalized.substr(start, dotPos - start);
        if (!isDigits(segment)) {
            valid = false;
            break;
        }
        ++segments;
        start = dotPos + 1;
    }

    if (!valid || segments < 1) {
        return {};
    }

    return lastSegment;
}

std::string determineChannelName(const utils::DeviceConfig& config) {
    const auto getConn = [&](const std::string& key) -> std::string {
        auto it = config.connection.find(key);
        if (it != config.connection.end()) {
            return it->second;
        }
        return {};
    };

    if (!config.name.empty()) {
        return config.name;
    }

    // TCP优先使用IP末段；无法获取时退回域名/端口或顺序号
    const std::string protoLower = toLower(config.protocol);
    if (protoLower == "modbus_tcp" || protoLower == "tcpmodbus") {
        std::string address = getConn("address");
        if (address.empty()) {
            address = getConn("ip");
        }
        if (address.empty()) {
            address = getConn("host");
        }
        if (!address.empty()) {
            const std::string ipLastOctet = extractIpLastOctet(address);
            if (!ipLastOctet.empty()) {
                return std::string("tcp-") + ipLastOctet;
            }
            std::string sanitized = sanitizeChannelToken(address);
            if (sanitized.empty()) {
                sanitized = replaceChar(address, '.', '-');
            }
            if (!sanitized.empty()) {
                return std::string("tcp-") + sanitized;
            }
        }
        std::string port = getConn("port");
        if (!port.empty()) {
            std::string sanitized = sanitizeChannelToken(port);
            if (!sanitized.empty()) {
                return std::string("tcp-") + sanitized;
            }
            return std::string("tcp-") + port;
        }
    }

    // RTU优先使用从站号，缺省时退回串口号
    if (protoLower == "modbus_rtu" || protoLower == "rtumodbus") {
        std::string slave = getConn("slave_id");
        if (slave.empty()) {
            slave = getConn("slave");
        }
        if (slave.empty()) {
            slave = getConn("id");
        }
        if (!slave.empty()) {
            std::string sanitized = sanitizeChannelToken(slave);
            if (!sanitized.empty()) {
                return std::string("rtu-") + sanitized;
            }
            return std::string("rtu-") + slave;
        }
        std::string port = getConn("port");
        if (!port.empty()) {
            std::string sanitized = sanitizeChannelToken(port);
            if (!sanitized.empty()) {
                return std::string("rtu-") + sanitized;
            }
            return std::string("rtu-") + port;
        }
    }

    if (!config.id.empty()) {
        return config.id;
    }
    return std::string("channel");
}

} // namespace

DataUploader::DataUploader() : running_(false), deviceManager_(nullptr) {
}

DataUploader::~DataUploader() {
    stop();
}

bool DataUploader::initialize() {
    LOG_INFO("Initializing data uploader");
    // 此处仅做轻量初始化，实际连接放在 start()
    return true;
}

void DataUploader::start() {
    if (running_) {
        return;
    }
    
    LOG_INFO("Starting data uploader");
    running_ = true;
    // 构建设备上报间隔映射
    buildDeviceUploadIntervals();
    // 先建立 MQTT 连接（按文档 v3 计算登录签名）
    if (!setupMqtt()) {
        LOG_MQTT_ERROR("MQTT setup failed, data upload will run without MQTT");
    }
    uploaderThread_ = std::thread(&DataUploader::uploadLoop, this);
}

void DataUploader::stop() {
    if (!running_) {
        return;
    }
    
    LOG_INFO("Stopping data uploader");
    running_ = false;
    
    if (uploaderThread_.joinable()) {
        uploaderThread_.join();
    }

    teardownMqtt();
}

void DataUploader::uploadLoop() {
    LOG_INFO("Data upload loop started");

    try {
        while (running_) {
            if (reauthRequested_.exchange(false)) {
                handleReauthentication();
            }

            if (reauthInProgress_) {
                std::this_thread::sleep_for(std::chrono::milliseconds(200));
                continue;
            }

            const long long t = nowMs();

        // 每设备-每模型按独立的 upload_interval 上报最新快照
        struct DueItem { std::string devId; std::string category; std::vector<device::DataValue> values; };
        std::vector<DueItem> dueItems;
        {
            std::lock_guard<std::mutex> lock(snapshotMutex_);
            for (auto& kv : snapshot_) {
                const std::string& devId = kv.first;
                const auto& data = kv.second;

                // 分类聚合
                std::unordered_map<std::string, std::vector<device::DataValue>> catMap;
                for (const auto& v : data) {
                    if (!v.valid) continue;
                    std::string cat = v.deviceCategory.empty() ? "unknown" : v.deviceCategory;
                    catMap[cat].push_back(v);
                }

                for (auto& ckv : catMap) {
                    const std::string rawCat = ckv.first;
                    const std::string normCat = normalizeCategory(rawCat);
                    const std::string keyRaw = devId + "|" + rawCat;
                    const std::string keyNorm = devId + "|" + normCat;
                    // 模型是否启用（未配置时视为启用）
                    auto itEn = modelEnabled_.find(keyRaw);
                    if (itEn == modelEnabled_.end()) {
                        itEn = modelEnabled_.find(keyNorm);
                    }
                    if (itEn != modelEnabled_.end() && !itEn->second) continue;

                    int sec = 0;
                    // 1) 模型级
                    auto itModel = deviceUploadIntervalSec_.find(keyRaw);
                    if (itModel == deviceUploadIntervalSec_.end()) {
                        itModel = deviceUploadIntervalSec_.find(keyNorm);
                    }
                    if (itModel != deviceUploadIntervalSec_.end() && itModel->second > 0) sec = itModel->second;
                    // 2) 设备级
                    if (sec <= 0) {
                        auto itDev = deviceUploadIntervalSec_.find(devId);
                        if (itDev != deviceUploadIntervalSec_.end() && itDev->second > 0) sec = itDev->second;
                    }
                    // 3) 系统默认（ms->s）
                    if (sec <= 0) {
                        const auto& sys = ems::utils::Config::getInstance().getSystem();
                        sec = std::max(1, sys.uploadInterval / 1000);
                    }

                    long long& nextTs = nextUploadMs_[keyRaw];
                    if (nextTs == 0) nextTs = t + (long long)sec * 1000;
                    if (mqttConnected_ && !reauthInProgress_ && t >= nextTs) {
                        dueItems.push_back(DueItem{devId, rawCat, ckv.second});
                        nextTs = t + (long long)sec * 1000;
                    }
                }
            }
        }

        for (auto& it : dueItems) {
            // 仅上报该模型类别的数据
            publishDataByCategory(it.devId, it.values);
        }

            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }

        LOG_INFO("Data upload loop stopped");
    } catch (const std::system_error& e) {
        std::fprintf(stderr, "[DataUploader::uploadLoop] system_error: %s\n", e.what());
        throw;
    }
}

void DataUploader::recordPublishResult(bool ok) {
    if (ok) {
        messagesSent_.fetch_add(1, std::memory_order_relaxed);
        const auto now = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        lastPublishMs_.store(now, std::memory_order_relaxed);
    } else {
        messagesFailed_.fetch_add(1, std::memory_order_relaxed);
    }
}

// ========================= 内部实现 =========================

std::string DataUploader::hexEncode(const unsigned char* data, size_t len) {
    static const char* hex = "0123456789abcdef";
    std::string out; out.resize(len * 2);
    for (size_t i = 0; i < len; ++i) {
        out[2*i]   = hex[(data[i] >> 4) & 0xF];
        out[2*i+1] = hex[data[i] & 0xF];
    }
    return out;
}

std::string DataUploader::hmacSha256Hex(const std::string& key, const std::string& content) {
    unsigned char md[EVP_MAX_MD_SIZE];
    unsigned int md_len = 0;
    HMAC(EVP_sha256(), key.data(), static_cast<int>(key.size()),
         reinterpret_cast<const unsigned char*>(content.data()), content.size(), md, &md_len);
    return hexEncode(md, md_len);
}

std::string DataUploader::md5Hex(const std::string& content) {
    unsigned char md[EVP_MAX_MD_SIZE];
    unsigned int md_len = 0;
    
    EVP_MD_CTX* ctx = EVP_MD_CTX_new();
    if (ctx == nullptr) return "";
    
    if (EVP_DigestInit_ex(ctx, EVP_md5(), nullptr) != 1 ||
        EVP_DigestUpdate(ctx, content.data(), content.size()) != 1 ||
        EVP_DigestFinal_ex(ctx, md, &md_len) != 1) {
        EVP_MD_CTX_free(ctx);
        return "";
    }
    
    EVP_MD_CTX_free(ctx);
    return hexEncode(md, md_len);
}

std::string DataUploader::randomDigits(int n) {
    static bool seeded = false;
    if (!seeded) {
        srand(static_cast<unsigned int>(time(nullptr)));
        seeded = true;
    }
    std::string s; s.reserve(n);
    for (int i = 0; i < n; ++i) s.push_back(static_cast<char>('0' + (rand() % 10)));
    return s;
}

std::string DataUploader::buildSignContent(const std::string& authType,
                                           const std::string& deviceSn,
                                           const std::string& deviceType,
                                           const std::string& random,
                                           int securemode) {
    // 按文档：字段名按字母序拼接，无分隔符
    // authType, deviceSn, deviceType, random, securemode
    std::ostringstream oss;
    oss << "authType" << authType
        << "deviceSn" << deviceSn
        << "deviceType" << deviceType
        << "random" << random
        << "securemode" << securemode;
    return oss.str();
}

std::string DataUploader::buildClientId(const std::string& deviceType,
                                        const std::string& deviceSn,
                                        const std::string& authType,
                                        const std::string& random,
                                        int securemode) {
    // 注意：虽然v3文档说登录时应该只用deviceSn，但实际测试发现平台还是期望带deviceType
    // 保持原有格式以确保兼容性
    std::ostringstream oss;
    oss << deviceType << "." << deviceSn
        << "|securemode=" << securemode
        << ",authType=" << authType
        << ",random=" << random
        << ",signMethod=hmacsha256";
    return oss.str();
}

bool DataUploader::readTextFile(const std::string& path, std::string& out) {
    std::ifstream ifs(path);
    if (!ifs.is_open()) return false;
    std::ostringstream ss; ss << ifs.rdbuf();
    out = ss.str();
    out.erase(std::remove(out.begin(), out.end(), '\r'), out.end());
    out.erase(std::remove(out.begin(), out.end(), '\n'), out.end());
    return true;
}

bool DataUploader::setupMqtt() {
    using ems::utils::Config; using ems::utils::MqttConfig; using ems::protocol::MqttClient; using ems::protocol::MqttQoS;
    auto& cfg = Config::getInstance();
    const MqttConfig& m = cfg.getMqtt();

    std::string host = m.host; int port = m.port > 0 ? m.port : 1883;
    bool useTls = m.useTls; int keepAlive = m.keepAlive > 0 ? m.keepAlive : 300;
    bool cleanSession = m.cleanSession; int qos = (m.qos >= 0 && m.qos <= 2) ? m.qos : 1;

    if (host.empty()) {
        LOG_MQTT_ERROR("MQTT host is empty in config/mqtt.conf");
        return false;
    }

    // 设备标识 - 优先使用硬件ID，如果配置中没有指定的话
    if (!m.deviceSn.empty() && m.deviceSn != "001" && m.deviceSn != "ems-client") {
        // 使用配置中的device_sn
        deviceSn_ = m.deviceSn;
    } else {
        // 从硬件获取唯一标识
        deviceSn_ = ems::utils::HardwareInfo::getDeviceId();
        LOG_INFO("Using hardware device ID: " << deviceSn_);
        if (cfg.saveMqttIdentity(deviceSn_, "")) {
            LOG_INFO("Persisted hardware device ID to config/mqtt.conf");
        }
    }
    deviceType_ = !m.deviceType.empty() ? m.deviceType : std::string("0");

    // 尝试加载已保存的设备密钥
    if (loadDeviceSecret()) {
        LOG_INFO("Found existing device secret, performing direct login");
        return setupMqttLogin();
    } else {
        LOG_INFO("No device secret found, performing dynamic registration");
        return performDynamicRegistration();
    }
}

void DataUploader::teardownMqtt() {
    std::unique_ptr<protocol::MqttClient> client;
    {
        if (!mqtt_) {
            mqttConnected_ = false;
            return;
        }
        client = std::move(mqtt_);
        mqttConnected_ = false;
    }
    LOG_INFO("Tearing down MQTT");
    if (client) {
        client->stopLoop();
        client->disconnect();
    }
}

// ====== 业务实现 ======

long long DataUploader::nowMs() {
    using namespace std::chrono;
    return duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count();
}

int DataUploader::extractSeq(const std::string& json, int defaultSeq) {
    // 简易解析，查找 "seq": <number>
    auto pos = json.find("\"seq\"");
    if (pos == std::string::npos) return defaultSeq;
    pos = json.find(':', pos);
    if (pos == std::string::npos) return defaultSeq;
    ++pos;
    while (pos < json.size() && isspace((unsigned char)json[pos])) ++pos;
    int sign = 1; if (pos < json.size() && json[pos] == '-') { sign = -1; ++pos; }
    long long val = 0; bool any=false;
    while (pos < json.size() && isdigit((unsigned char)json[pos])) { any=true; val = val*10 + (json[pos]-'0'); ++pos; }
    if (!any) return defaultSeq; return (int)(val * sign);
}

void DataUploader::publishLogin() {
    if (!mqtt_ || !mqttConnected_) return;
    // 构造登录消息（文档v3）：cmdId=101
    // deviceType 采用数字输出
    int devTypeNum = 0; try { devTypeNum = std::stoi(deviceType_); } catch (...) {}
    long long ts = nowMs();
    int randomNum = 12123; // 可以使用固定值或随机数
    
    // 计算签名: MD5(gatewayId + 'Sunfly_Aiot_Platform' + random)
    std::string gwId = std::string("ems-") + deviceSn_; // 添加ems-前缀
    std::string signContent = gwId + "Sunfly_Aiot_Platform" + std::to_string(randomNum);
    std::string sign = md5Hex(signContent);
    
    std::ostringstream ss;
    ss << "{\"cmdId\":101,\"seq\":" << (ts % 1000000)
       << ",\"gwId\":\"" << gwId << "\",\"errorCode\":0,\"data\":{"
       << "\"sign\":\"" << sign << "\","
       << "\"random\":" << randomNum << ","
       << "\"deviceType\":" << devTypeNum << ","
       << "\"hardwareVersion\":\"\","
       << "\"softwareVersion\":\"\","
       << "\"tplVersion\":\"\","
       << "\"gatewayMac\":\"\","
       << "\"remoteIp\":\"\","
       << "\"remotePort\":0,"
       << "\"cellCode\":\"\","
       << "\"floorCode\":\"\","
       << "\"houseCode\":\"\","
       << "\"tenantId\":\"\"}}";
    const std::string payload = ss.str();
    
    LOG_MQTT_SEND("=== MQTT LOGIN MESSAGE SENDING ===");
    LOG_MQTT_SEND("Topic: " << topicLogin_);
    LOG_MQTT_SEND("Payload Length: " << payload.size());
    LOG_MQTT_SEND("Payload Content: " << payload);
    LOG_MQTT_SEND("==================================");
    
    bool ok = mqtt_->publish(topicLogin_, payload, ems::protocol::MqttQoS::QOS_1, false);
    recordPublishResult(ok);
    if (ok) LOG_MQTT_SEND("MQTT login publish ok: topic=" << topicLogin_);
    else    LOG_MQTT_ERROR("MQTT login publish fail: topic=" << topicLogin_ << ", err=" << mqtt_->getLastError());
}

void DataUploader::publishDataSample() {
    if (reauthInProgress_) return;
    if (!mqtt_ || !mqttConnected_) return;
    // 去除 _test_ping：仅发送时间与设备标识（示例心跳结构）
    long long epoch = (long long)std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count();
    std::ostringstream ss;
    // 与数据上报格式对齐：添加 pId，dSn 使用与同步一致的子设备命名（默认pcs）
    std::string sampleDeviceId;
    std::string sampleCategory = "pcs";
    int productId = 10134;
    if (!modelDeviceInfo_.empty()) {
        const auto& info = modelDeviceInfo_.begin()->second;
        sampleDeviceId = info.deviceId;
        sampleCategory = info.category;
        if (info.deviceType > 0) {
            productId = info.deviceType;
        } else {
            productId = resolveModelDeviceType(info.deviceId, info.category);
        }
    }
    ss << "{\"cmdId\":312,\"gwId\":\"ems-" << deviceSn_ << "\",\"data\":{\"time\":" << epoch
       << ",\"pId\":" << productId
       << ",\"dSn\":\"" << makeSubDeviceSn(sampleDeviceId.empty()? "default" : sampleDeviceId, sampleCategory) << "\"}}"; 
    const std::string payload = ss.str();
    
    LOG_MQTT_SEND("=== MQTT DATA MESSAGE SENDING ===");
    LOG_MQTT_SEND("Topic: " << topicDataUpload_);
    LOG_MQTT_SEND("Payload Length: " << payload.size());
    LOG_MQTT_SEND("Payload Content: " << payload);
    LOG_MQTT_SEND("=================================");
    
    bool ok = mqtt_->publish(topicDataUpload_, payload, ems::protocol::MqttQoS::QOS_1, false);
    recordPublishResult(ok);
    if (ok) LOG_MQTT_SEND("MQTT data publish ok: topic=" << topicDataUpload_);
    else    LOG_MQTT_ERROR("MQTT data publish fail: topic=" << topicDataUpload_ << ", err=" << mqtt_->getLastError());
}

void DataUploader::publishDataByCategory(const std::string& deviceId, const std::vector<device::DataValue>& dataValues) {
    if (reauthInProgress_) return;
    if (!mqtt_ || !mqttConnected_) return;

    // 按device_category分类数据
    std::map<std::string, std::vector<device::DataValue>> categorizedData;
    for (const auto& data : dataValues) {
        if (!data.valid) continue;
        std::string category = data.deviceCategory.empty() ? "unknown" : data.deviceCategory;
        categorizedData[category].push_back(data);
    }

    // 为每个category生成独立的MQTT消息
    for (const auto& categoryPair : categorizedData) {
        const std::string& category = categoryPair.first;
        const std::vector<device::DataValue>& values = categoryPair.second;

        if (values.empty()) continue;

        long long epoch = std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();

        std::ostringstream ss;
        ss << "{";
        ss << "\"cmdId\":312,";
        ss << "\"seq\":" << (rand() % 10000) << ",";
        ss << "\"gwId\":\"ems-" << deviceSn_ << "\",";
        ss << "\"data\":{";
        ss << "\"time\":" << epoch << ",";
        // pId: 按模型配置或默认映射
        int productId = resolveModelDeviceType(deviceId, category);
        if (productId <= 0) {
            std::string catLower = normalizeCategory(category);
            if (catLower == "bms") productId = 10133;
            else if (catLower == "alarm") productId = 10132;
            else productId = 10134; // 默认PCS
        }
        ss << "\"pId\":" << productId << ",";
        // 上报设备SN按同步时的命名（gwId-类别），确保与v3规范一致
        ss << "\"dSn\":\"" << makeSubDeviceSn(deviceId, category) << "\",";
        ss << "\"events\":[";

        // 添加该类别的所有数据点
        bool first = true;
        for (const auto& value : values) {
            if (!first) ss << ",";
            ss << "{";
            // key 追加 _mete 后缀（如已包含则不重复追加）
            {
                std::string keyOut = value.key;
                if (!(keyOut.size() >= 5 && keyOut.compare(keyOut.size()-5, 5, "_mete") == 0)) {
                    keyOut += "_mete";
                }
                ss << "\"key\":\"" << keyOut << "\",";
            }
            // value 统一转为字符串类型
            {
                std::ostringstream vs; vs << value.value;
                ss << "\"value\":\"" << vs.str() << "\"";
            }
            ss << "}";
            first = false;
        }

        ss << "]}";
        ss << "}";

        const std::string payload = ss.str();

        LOG_INFO("MQTT分类上报 - " << category << " 类别数据，包含 " << values.size() << " 个数据点");
        LOG_DEBUG("MQTT Payload: " << payload);

        bool ok = mqtt_->publish(topicDataUpload_, payload, ems::protocol::MqttQoS::QOS_1, false);
        recordPublishResult(ok);
        if (ok) {
            LOG_INFO("MQTT " << category << " 数据上报成功");
        } else {
            LOG_ERROR("MQTT " << category << " 数据上报失败: " << mqtt_->getLastError());
        }
    }
}

void DataUploader::handleIncomingMessage(const std::string& topic, const std::string& payload) {
    // 简单按路径匹配
    if (topic.find("/sync") != std::string::npos) {
        handleSyncRequest(topic, payload);
        return;
    }
    if (topic.find("/func") != std::string::npos) {
        handleFuncRequest(topic, payload);
        return;
    }
    if (topic.find("/login_rsp") != std::string::npos) {
        LOG_INFO("MQTT login_rsp received");
        return;
    }
}

void DataUploader::handleSyncRequest(const std::string& topic, const std::string& payload) {
    LOG_MQTT_RECV("=== MQTT SYNC REQUEST RECEIVED ===");
    LOG_MQTT_RECV("Topic: " << topic);
    LOG_MQTT_RECV("Payload Length: " << payload.size());
    LOG_MQTT_RECV("Payload Content: " << payload);
    LOG_MQTT_RECV("====================================");

    if (reauthInProgress_) return;
    (void)topic;
    int seq = extractSeq(payload, 0);
    // 构造设备同步响应，根据配置中的 models 定义返回子设备列表
    // 同步响应cmdId=302
    std::ostringstream ss;
    ss << "{\"cmdId\":302,\"seq\":" << seq << ",\"gwId\":\"ems-" << deviceSn_ << "\",\"errorCode\":0,\"data\":[";

    std::vector<ModelDeviceInfo> enabledModels;
    enabledModels.reserve(modelDeviceInfo_.size());
    for (const auto& kv : modelDeviceInfo_) {
        if (!kv.second.enabled) continue;
        if (!kv.second.deviceSn.empty() || !kv.second.deviceName.empty()) {
            enabledModels.push_back(kv.second);
        }
    }

    if (enabledModels.empty()) {
        LOG_INFO("No configured models available; sending empty MQTT device sync response");
    }

    bool first = true;
    for (const auto& info : enabledModels) {
        if (!first) ss << ","; else first = false;
        int onlineState = info.enabled ? 1 : 0;
        if (deviceManager_ && info.enabled) {
            onlineState = deviceManager_->getConnectedDeviceCount() > 0 ? 1 : 0;
        }
        const int type = info.deviceType > 0 ? info.deviceType : resolveModelDeviceType(info.deviceId, info.category);
        const std::string subSn = !info.deviceSn.empty() ? info.deviceSn : makeSubDeviceSn(info.deviceId, info.category);
        const std::string deviceName = resolveModelDeviceName(info);

        ss << "{";
        ss << "\"deviceType\":" << (type > 0 ? type : 0) << ",";
        ss << "\"deviceName\":\"" << deviceName << "\",";
        ss << "\"deviceSn\":\"" << subSn << "\",";
        ss << "\"onlineState\":" << onlineState << ",";
        ss << "\"subDeviceCount\":0";
        ss << "}";
    }

    ss << "]}";
    const std::string resp = ss.str();
    
    LOG_MQTT_SEND("=== MQTT SYNC RESPONSE SENDING ===");
    LOG_MQTT_SEND("Topic: " << topicDeviceSyncResp_);
    LOG_MQTT_SEND("Response Length: " << resp.size());
    LOG_MQTT_SEND("Response Content: " << resp);
    LOG_MQTT_SEND("==================================");
    
    if (mqtt_ && mqttConnected_ && !reauthInProgress_) {
        bool ok = mqtt_->publish(topicDeviceSyncResp_, resp, ems::protocol::MqttQoS::QOS_1, false);
        recordPublishResult(ok);
        if (ok) LOG_MQTT_SEND("MQTT sync resp ok: topic=" << topicDeviceSyncResp_);
        else    LOG_MQTT_ERROR("MQTT sync resp fail: topic=" << topicDeviceSyncResp_ << ", err=" << mqtt_->getLastError());
    }
}

void DataUploader::handleFuncRequest(const std::string& topic, const std::string& payload) {
    LOG_MQTT_RECV("=== MQTT FUNCTION REQUEST RECEIVED ===");
    LOG_MQTT_RECV("Topic: " << topic);
    LOG_MQTT_RECV("Payload Length: " << payload.size());
    LOG_MQTT_RECV("Payload Content: " << payload);
    LOG_MQTT_RECV("======================================");
    
    (void)topic;
    int seq = extractSeq(payload, 0);
    // 简化：直接应答 ok
    if (reauthInProgress_) return;
    std::ostringstream ss;
    // 文档约定：控制响应 cmdId=404
    ss << "{\"cmdId\":404,\"seq\":" << seq << ",\"gwId\":\"ems-" << deviceSn_ << "\",\"errorCode\":0,\"data\":{\"result\":\"ok\"}}";
    const std::string resp = ss.str();
    
    LOG_MQTT_SEND("=== MQTT FUNCTION RESPONSE SENDING ===");
    LOG_MQTT_SEND("Topic: " << topicCommandResp_);
    LOG_MQTT_SEND("Response Length: " << resp.size());
    LOG_MQTT_SEND("Response Content: " << resp);
    LOG_MQTT_SEND("======================================");
    
    if (mqtt_ && mqttConnected_ && !reauthInProgress_) {
        bool ok = mqtt_->publish(topicCommandResp_, resp, ems::protocol::MqttQoS::QOS_1, false);
        recordPublishResult(ok);
        if (ok) LOG_MQTT_SEND("MQTT func resp ok: topic=" << topicCommandResp_);
        else    LOG_MQTT_ERROR("MQTT func resp fail: topic=" << topicCommandResp_ << ", err=" << mqtt_->getLastError());
    }
}

// ================== 动态注册功能 ==================

bool DataUploader::loadDeviceSecret() {
    std::string secretPath = std::string("./config/device_secret_") + deviceSn_ + ".txt";
    return readTextFile(secretPath, deviceSecret_) && !deviceSecret_.empty();
}

bool DataUploader::saveDeviceSecret(const std::string& secret) {
    std::string secretPath = std::string("./config/device_secret_") + deviceSn_ + ".txt";
    std::ofstream file(secretPath);
    if (!file.is_open()) {
        LOG_ERROR("Failed to save device secret to: " << secretPath);
        return false;
    }
    file << secret;
    file.close();
    deviceSecret_ = secret;
    LOG_INFO("Device secret saved successfully");
    return true;
}

bool DataUploader::removeDeviceSecret() {
    std::string secretPath = std::string("./config/device_secret_") + deviceSn_ + ".txt";
    deviceSecret_.clear();
    std::error_code ec;
    if (std::remove(secretPath.c_str()) == 0) {
        LOG_INFO("Removed saved device secret: " << secretPath);
        return true;
    } else {
        // 文件不存在也视为成功
        LOG_WARN("Device secret file not found or cannot be removed: " << secretPath);
        return false;
    }
}

bool DataUploader::performDynamicRegistration() {
    using ems::utils::Config; using ems::utils::MqttConfig; using ems::protocol::MqttClient; using ems::protocol::MqttQoS;
    auto& cfg = Config::getInstance();
    const MqttConfig& m = cfg.getMqtt();
    
    std::string host = m.host; int port = m.port > 0 ? m.port : 1883;
    bool useTls = m.useTls; int keepAlive = m.keepAlive > 0 ? m.keepAlive : 300;
    bool cleanSession = m.cleanSession;
    
    // 动态注册参数
    const int securemode = useTls ? 1 : 2;
    const std::string random = randomDigits(8);
    // 动态注册时，签名计算使用带ems-前缀的deviceSn
    const std::string emsDeviceSn = std::string("ems-") + deviceSn_;
    const std::string content = buildSignContent("register", emsDeviceSn, deviceType_, random, securemode);
    std::string username = emsDeviceSn + "&" + deviceType_;
    std::string clientId = buildClientId(deviceType_, emsDeviceSn, "register", random, securemode);
    
    // 使用device_type_secret进行注册
    std::string password;
    if (!m.deviceTypeSecret.empty()) {
        password = hmacSha256Hex(m.deviceTypeSecret, content);
        LOG_INFO("Using device_type_secret for dynamic registration");
    } else {
        LOG_ERROR("device_type_secret not configured, cannot perform dynamic registration");
        return false;
    }
    
    // 创建MQTT客户端用于注册
    mqtt_.reset(new MqttClient());
    mqtt_->setKeepAlive(keepAlive);
    mqtt_->setCleanSession(cleanSession);
    mqtt_->setAutoReconnect(false);  // 注册完成后需要断开
    
    if (!mqtt_->initialize(clientId, host, port)) {
        LOG_MQTT_ERROR("Initialize MQTT client for registration failed");
        mqtt_.reset();
        return false;
    }
    
    if (!mqtt_->setCredentials(username, password)) {
        LOG_MQTT_ERROR("Set MQTT credentials for registration failed: " << mqtt_->getLastError());
        mqtt_.reset();
        return false;
    }
    
    if (useTls) {
        if (!m.caFile.empty() || !m.certFile.empty() || !m.keyFile.empty()) {
            if (!mqtt_->setTLS(m.caFile, m.certFile, m.keyFile)) {
                LOG_MQTT_ERROR("Set TLS for registration failed: " << mqtt_->getLastError());
                mqtt_.reset();
                return false;
            }
        }
    }
    
    // 设置连接和消息回调
    registrationState_ = RegistrationState::REGISTERING;
    mqtt_->setConnectCallback([this](bool ok){
        mqttConnected_ = ok;
        if (ok) {
            LOG_MQTT_CONN("MQTT registration connection established");
            // 订阅注册响应主题
            std::string regTopic = "/ext/ems-" + deviceSn_ + "/register";
            mqtt_->subscribe(regTopic, MqttQoS::QOS_1);
            LOG_INFO("Subscribed to registration topic: " << regTopic);
        } else {
            // 只有在没有成功注册的情况下才认为连接失败
            if (registrationState_ != RegistrationState::REGISTERED) {
                LOG_MQTT_ERROR("MQTT registration connection failed");
                registrationState_ = RegistrationState::FAILED;
            } else {
                LOG_INFO("MQTT registration connection closed after successful registration");
            }
        }
    });
    
    mqtt_->setMessageCallback([this](const std::string& topic, const std::string& payload){
        LOG_MQTT_RECV("=== MQTT REGISTRATION MESSAGE RECEIVED ===");
        LOG_MQTT_RECV("Topic: " << topic);
        LOG_MQTT_RECV("Payload Length: " << payload.size());
        LOG_MQTT_RECV("Payload Content: " << payload);
        LOG_MQTT_RECV("==========================================");
        handleRegistrationResponse(topic, payload);
    });
    
    if (!mqtt_->connect()) {
        LOG_MQTT_ERROR("MQTT registration connect failed: " << mqtt_->getLastError());
        mqtt_.reset();
        registrationState_ = RegistrationState::FAILED;
        return false;
    }
    
    // 打印动态注册请求信息
    LOG_INFO("=== MQTT DYNAMIC REGISTRATION REQUEST ===");
    LOG_INFO("ClientId: " << clientId);
    LOG_INFO("Username: " << username);
    LOG_INFO("DeviceId: " << deviceSn_);
    LOG_INFO("DeviceType: " << deviceType_);
    LOG_INFO("AuthType: register");
    LOG_INFO("SecureMode: " << securemode);
    LOG_INFO("Random: " << random);
    LOG_INFO("=========================================");
    
    mqtt_->startLoop();
    LOG_INFO("MQTT registration loop started, waiting for device secret...");
    
    // 等待注册完成（最多30秒），但在收到DeviceSecret后立即退出
    for (int i = 0; i < 30 && registrationState_ == RegistrationState::REGISTERING; i++) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    
    // 按照MQTT文档要求：收到DeviceSecret后断开连接
    LOG_INFO("Stopping MQTT registration connection");
    mqtt_->stopLoop();
    mqtt_->disconnect();
    mqtt_.reset();
    
    if (registrationState_ == RegistrationState::REGISTERED) {
        LOG_INFO("Dynamic registration completed successfully");
        // 现在使用获取到的device_secret进行正常登录
        return setupMqttLogin();
    } else {
        LOG_ERROR("Dynamic registration failed");
        return false;
    }
}

void DataUploader::handleRegistrationResponse(const std::string& topic, const std::string& payload) {
    // 检查是否是注册响应主题
    std::string expectedTopic = "/ext/ems-" + deviceSn_ + "/register";
    if (topic != expectedTopic) {
        return;
    }
    
    try {
        // 简单的JSON解析（假设格式正确）
        size_t pos = payload.find("\"deviceSecret\"");
        if (pos == std::string::npos) {
            LOG_ERROR("Invalid registration response: missing deviceSecret");
            registrationState_ = RegistrationState::FAILED;
            return;
        }
        
        pos = payload.find(":", pos);
        if (pos == std::string::npos) {
            LOG_ERROR("Invalid registration response: malformed deviceSecret");
            registrationState_ = RegistrationState::FAILED;
            return;
        }
        
        pos = payload.find("\"", pos + 1);
        if (pos == std::string::npos) {
            LOG_ERROR("Invalid registration response: deviceSecret not quoted");
            registrationState_ = RegistrationState::FAILED;
            return;
        }
        
        size_t end = payload.find("\"", pos + 1);
        if (end == std::string::npos) {
            LOG_ERROR("Invalid registration response: deviceSecret quote not closed");
            registrationState_ = RegistrationState::FAILED;
            return;
        }
        
        std::string deviceSecret = payload.substr(pos + 1, end - pos - 1);
        if (deviceSecret.empty()) {
            LOG_ERROR("Received empty device secret");
            registrationState_ = RegistrationState::FAILED;
            return;
        }
        
        LOG_INFO("Received device secret from registration");
        if (saveDeviceSecret(deviceSecret)) {
            registrationState_ = RegistrationState::REGISTERED;
            LOG_INFO("Device secret saved, will disconnect and re-login with device secret");
        } else {
            registrationState_ = RegistrationState::FAILED;
        }
        
    } catch (const std::exception& e) {
        LOG_ERROR("Exception parsing registration response: " << e.what());
        registrationState_ = RegistrationState::FAILED;
    }
}

bool DataUploader::setupMqttLogin() {
    using ems::utils::Config; using ems::utils::MqttConfig; using ems::protocol::MqttClient; using ems::protocol::MqttQoS;
    auto& cfg = Config::getInstance();
    const MqttConfig& m = cfg.getMqtt();
    
    std::string host = m.host; int port = m.port > 0 ? m.port : 1883;
    bool useTls = m.useTls; int keepAlive = m.keepAlive > 0 ? m.keepAlive : 300;
    bool cleanSession = m.cleanSession; int qos = (m.qos >= 0 && m.qos <= 2) ? m.qos : 1;
    
    // 使用device_secret进行登录
    const int securemode = useTls ? 1 : 2;
    const std::string random = randomDigits(8);
    // 登录时，签名计算使用带ems-前缀的deviceSn（与gwId保持一致）
    const std::string emsDeviceSn = std::string("ems-") + deviceSn_;
    const std::string content = buildSignContent("login", emsDeviceSn, deviceType_, random, securemode);
    std::string username = emsDeviceSn + "&" + deviceType_;
    std::string clientId = buildClientId(deviceType_, emsDeviceSn, "login", random, securemode);
    std::string password = hmacSha256Hex(deviceSecret_, content);
    
    mqtt_.reset(new MqttClient());
    mqtt_->setKeepAlive(keepAlive);
    mqtt_->setCleanSession(cleanSession);
    mqtt_->setAutoReconnect(true);
    
    if (!mqtt_->initialize(clientId, host, port)) {
        LOG_MQTT_ERROR("Initialize MQTT client for login failed");
        mqtt_.reset();
        return false;
    }
    
    if (!mqtt_->setCredentials(username, password)) {
        LOG_MQTT_ERROR("Set MQTT credentials for login failed: " << mqtt_->getLastError());
        mqtt_.reset();
        return false;
    }
    
    if (useTls) {
        if (!m.caFile.empty() || !m.certFile.empty() || !m.keyFile.empty()) {
            if (!mqtt_->setTLS(m.caFile, m.certFile, m.keyFile)) {
                LOG_MQTT_ERROR("Set TLS for login failed: " << mqtt_->getLastError());
                mqtt_.reset();
                return false;
            }
        }
    }
    
    mqtt_->setConnectCallback([this](bool ok){
        mqttConnected_ = ok;
        if (ok) {
            LOG_MQTT_CONN("MQTT login connected successfully"); 
        } else {
            // 读取最后错误，判断是否为认证失败
            std::string err = mqtt_ ? mqtt_->getLastError() : std::string();
            std::string el = err; std::transform(el.begin(), el.end(), el.begin(), [](unsigned char c){ return std::tolower(c); });
            bool notAuth = (el.find("not authorized") != std::string::npos) || (el.find("not authorised") != std::string::npos) || (el.find("authoris") != std::string::npos);
            if (notAuth && !reauthAttempted_.exchange(true)) {
                LOG_MQTT_ERROR("MQTT authentication failed (not authorized). Re-registration scheduled.");
                reauthRequested_.store(true);
            } else {
                LOG_MQTT_CONN("MQTT login disconnected");
            }
        }
    });
    
    mqtt_->setMessageCallback([this](const std::string& topic, const std::string& payload){
        LOG_MQTT_RECV("=== MQTT LOGIN MESSAGE RECEIVED ===");
        LOG_MQTT_RECV("Topic: " << topic);
        LOG_MQTT_RECV("Payload Length: " << payload.size());
        LOG_MQTT_RECV("Payload Content: " << payload);
        LOG_MQTT_RECV("===================================");
        handleIncomingMessage(topic, payload);
    });
    
    if (!mqtt_->connect()) {
        LOG_MQTT_ERROR("MQTT login connect failed: " << mqtt_->getLastError());
        mqtt_.reset();
        return false;
    }
    
    mqtt_->startLoop();
    LOG_INFO("MQTT login loop started");
    
    // 订阅下行主题
    std::string downTopic = std::string("down/gateway/ems-") + deviceSn_ + "/#";
    mqtt_->subscribe(downTopic, static_cast<MqttQoS>(qos));
    LOG_INFO("Subscribed: " << downTopic);
    
    // 主题映射
    topicLogin_         = m.topics.count("login")           ? m.topics.at("login")           : std::string("up/gateway/login");
    topicDataUpload_    = m.topics.count("data_upload")     ? m.topics.at("data_upload")     : std::string("up/gateway/report/data");
    topicAlarmUpload_   = m.topics.count("alarm_upload")    ? m.topics.at("alarm_upload")    : std::string("up/gateway/report/alarm");
    topicAlarmClear_    = m.topics.count("alarm_clear")     ? m.topics.at("alarm_clear")     : std::string("up/gateway/report/alarm_clear");
    topicDeviceSyncResp_= m.topics.count("device_sync")     ? m.topics.at("device_sync")     : std::string("up/gateway/sync/res");
    topicCommandResp_   = m.topics.count("command_response")? m.topics.at("command_response"): std::string("up/gateway/func/res");
    
    // 连接成功后发送登录消息
    publishLogin();
    return true;
}

// ============== 新增实现 ==============
void DataUploader::updateSnapshot(const std::string& deviceId, const std::vector<device::DataValue>& dataValues) {
    std::lock_guard<std::mutex> lock(snapshotMutex_);
    snapshot_[deviceId] = dataValues; // 覆盖最新快照
}

void DataUploader::clearSnapshot(const std::string& deviceId) {
    std::lock_guard<std::mutex> lock(snapshotMutex_);
    snapshot_.erase(deviceId);
    const std::string prefix = deviceId + "|";
    for (auto it = nextUploadMs_.begin(); it != nextUploadMs_.end(); ) {
        if (it->first.rfind(prefix, 0) == 0) {
            it = nextUploadMs_.erase(it);
        } else {
            ++it;
        }
    }
}

void DataUploader::rebuildDeviceIntervals() {
    std::lock_guard<std::mutex> lock(snapshotMutex_);
    buildDeviceUploadIntervals();

    std::unordered_set<std::string> validDevices;
    std::unordered_set<std::string> validIntervalKeys;
    const auto& devices = ems::utils::Config::getInstance().getDevices();
    for (const auto& d : devices) {
        validDevices.insert(d.id);
    }

    for (const auto& entry : deviceUploadIntervalSec_) {
        validIntervalKeys.insert(entry.first);
    }

    for (auto it = snapshot_.begin(); it != snapshot_.end(); ) {
        if (validDevices.find(it->first) == validDevices.end()) {
            it = snapshot_.erase(it);
        } else {
            ++it;
        }
    }

    const long long now = nowMs();
    for (auto it = nextUploadMs_.begin(); it != nextUploadMs_.end(); ) {
        if (validIntervalKeys.find(it->first) == validIntervalKeys.end()) {
            it = nextUploadMs_.erase(it);
        } else {
            ++it;
        }
    }
    for (const auto& key : validIntervalKeys) {
        nextUploadMs_[key] = now;
    }
}

bool DataUploader::tryGetCachedValue(const std::string& deviceId,
                                     const std::string& key,
                                     device::DataValue& out) const {
    std::lock_guard<std::mutex> lock(snapshotMutex_);
    auto it = snapshot_.find(deviceId);
    if (it == snapshot_.end()) {
        return false;
    }
    const auto& values = it->second;
    for (const auto& v : values) {
        if (v.key == key) {
            out = v;
            return v.valid;
        }
    }
    return false;
}

std::unordered_map<std::string, std::vector<device::DataValue>> DataUploader::getSnapshotCopy() const {
    std::lock_guard<std::mutex> lock(snapshotMutex_);
    return snapshot_;
}

void DataUploader::buildDeviceUploadIntervals() {
    deviceUploadIntervalSec_.clear();
    modelEnabled_.clear();
    modelDeviceInfo_.clear();
    modelAlias_.clear();
    auto& cfg = ems::utils::Config::getInstance();
    const auto& devs = cfg.getDevices();
    for (const auto& d : devs) {
        // 设备级默认（作为兜底）
        if (d.uploadInterval > 0) {
            deviceUploadIntervalSec_[d.id] = d.uploadInterval; // 秒
        }
        // 模型级配置
        for (const auto& mkv : d.models) {
            const std::string& model = mkv.first; // e.g. bms-1/pcs-1
            const auto& mc = mkv.second;
            const std::string normCat = normalizeCategory(model);
            const std::string rawKey = d.id + "|" + model;
            const std::string normKey = d.id + "|" + normCat;
            modelEnabled_[rawKey] = mc.enabled;
            modelEnabled_[normKey] = mc.enabled;
            if (mc.uploadIntervalSec > 0) {
                deviceUploadIntervalSec_[rawKey] = mc.uploadIntervalSec; // 秒
                deviceUploadIntervalSec_[normKey] = mc.uploadIntervalSec; // 秒
            }

            ModelDeviceInfo info;
            info.deviceId = d.id;
            info.category = normCat;
            info.rawCategory = model;
            info.enabled = mc.enabled;
            info.deviceSn = mc.deviceSn;
            info.deviceType = mc.deviceType > 0 ? mc.deviceType : 0;
            info.deviceName = mc.deviceName;
            if (info.deviceType == 0) {
                if (normCat == "bms") info.deviceType = 10133;
                else if (normCat == "pcs") info.deviceType = 10134;
                else if (normCat == "alarm") info.deviceType = 10132;
            }
            if (info.deviceName.empty()) {
                info.deviceName = model;
            }
            info.channelName = determineChannelName(d);
            modelDeviceInfo_[rawKey] = info;
            if (normKey != rawKey) {
                modelAlias_[normKey] = rawKey;
            }
        }
    }
}

void DataUploader::handleReauthentication() {
    if (reauthInProgress_) {
        return;
    }
    reauthInProgress_ = true;
    LOG_INFO("Starting MQTT re-authentication workflow");
    try {
        teardownMqtt();
        removeDeviceSecret();
        if (!performDynamicRegistration()) {
            LOG_MQTT_ERROR("Re-authentication via dynamic registration failed");
        } else {
            LOG_INFO("Re-authentication succeeded via dynamic registration");
            reauthAttempted_.store(false);
        }
    } catch (const std::exception& ex) {
        LOG_MQTT_ERROR(std::string("Exception during re-authentication: ") + ex.what());
    }
    reauthInProgress_ = false;
}

std::string DataUploader::normalizeCategory(const std::string& category) {
    std::string lower = category;
    std::transform(lower.begin(), lower.end(), lower.begin(), [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
    // 若末尾存在 "-<数字>"，视为序号后缀，截断成基础类别
    auto pos = lower.find_last_of('-');
    if (pos != std::string::npos && pos + 1 < lower.size()) {
        bool numericSuffix = true;
        for (size_t i = pos + 1; i < lower.size(); ++i) {
            if (!std::isdigit(static_cast<unsigned char>(lower[i]))) {
                numericSuffix = false;
                break;
            }
        }
        if (numericSuffix) {
            lower = lower.substr(0, pos);
        }
    }
    return lower;
}

const DataUploader::ModelDeviceInfo* DataUploader::findModelInfo(const std::string& deviceId, const std::string& category) const {
    const std::string rawKey = deviceId + "|" + category;
    auto it = modelDeviceInfo_.find(rawKey);
    if (it != modelDeviceInfo_.end()) {
        return &it->second;
    }
    const std::string key = deviceId + "|" + normalizeCategory(category);
    auto aliasIt = modelAlias_.find(key);
    if (aliasIt != modelAlias_.end()) {
        auto base = modelDeviceInfo_.find(aliasIt->second);
        if (base != modelDeviceInfo_.end()) {
            return &base->second;
        }
    }
    it = modelDeviceInfo_.find(key);
    if (it != modelDeviceInfo_.end()) {
        return &it->second;
    }
    return nullptr;
}

int DataUploader::resolveModelDeviceType(const std::string& deviceId, const std::string& category) const {
    const auto* info = findModelInfo(deviceId, category);
    if (info && info->deviceType > 0) {
        return info->deviceType;
    }
    const std::string norm = normalizeCategory(category);
    if (norm == "bms") return 10133;
    if (norm == "pcs") return 10134;
    if (norm == "alarm") return 10132;
    return 0;
}

std::string DataUploader::resolveModelDeviceName(const ModelDeviceInfo& info) const {
    if (!info.deviceName.empty()) return info.deviceName;
    if (!info.rawCategory.empty()) return info.rawCategory;
    return info.category;
}

std::string DataUploader::makeSubDeviceSn(const std::string& deviceId, const std::string& category) const {
    const auto* info = findModelInfo(deviceId, category);
    if (info && !info->deviceSn.empty()) {
        return info->deviceSn;
    }
    // 兜底：通道名 + 网关ID + 模型，确保与前端策略一致
    const std::string channel = (info && !info->channelName.empty()) ? info->channelName : (!deviceId.empty() ? deviceId : std::string("channel"));
    const std::string gwId = sanitizeGatewayId();
    const std::string model = (info && !info->rawCategory.empty()) ? info->rawCategory : category;

    if (channel.empty()) {
        return gwId + "-" + model;
    }

    if (model.empty()) {
        return channel + "-" + gwId;
    }

    return channel + "-" + gwId + "-" + model;
}

std::string DataUploader::sanitizeGatewayId() const {
    if (deviceSn_.empty()) {
        return std::string("gateway");
    }
    // MQTT登录使用的deviceSn可能包含分隔符，此处统一清洗供子设备SN复用
    std::string candidate = sanitizeChannelToken(deviceSn_);
    if (candidate.empty()) {
        candidate = deviceSn_;
    }
    const std::string prefix = "ems-";
    if (candidate.size() > prefix.size() && candidate.rfind(prefix, 0) == 0) {
        candidate = candidate.substr(prefix.size());
    }
    if (candidate.empty()) {
        return std::string("gateway");
    }
    return candidate;
}

}  // namespace service
}  // namespace ems
